package com.distribute.framework.plugin;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.distribute.framework.config.DatabaseNames;
import com.alibaba.fastjson.JSON;
import com.distribute.framework.core.lang.Message;
import com.distribute.framework.core.sql.redis.RedisSupport;
import com.distribute.framework.core.web.helper.Utils;


public class Notify {
	
	private static final String defaultFlag = "framework";
	private static Map<String,String> topices = new HashMap();
	private static Long seconds = 10l;
	static{
		new Thread(){
			public void run(){
				while(true){
					try{
						synchronized(topices){
							for(Map.Entry<String, String> entry:topices.entrySet()){
								RedisSupport.getRedis(Cache.defaultRedisFlag).setValue("Notify", entry.getKey(), entry.getValue(), seconds.intValue()*2);
							}
						}
					}catch(Exception ex){
						ex.printStackTrace();
					}
					try{
						Thread.sleep(seconds*1000);
					}catch(Exception ex){
						ex.printStackTrace();
					}
				}
			}
		}.start();
	}
	/**
	 * 获取
	 * @param flag 无用
	 * @param topic
	 * @return
	 */
	public static String getSubscribe(String flag,String topic){
		return RedisSupport.getRedis(Cache.defaultRedisFlag).getValue("Notify", topic);
	}
	public static void addSubscribe(String topic){
		topices.put(topic, topic);
	}
	
	static Map<String,BaseMQ> mqs = new HashMap();
	
	public static BaseMQ getMQ(String key){
		if(mqs.containsKey(key)){
			return mqs.get(key);
		}
		try {
			BaseMQ mq = (BaseMQ) Class.forName("com.distribute.framework.plugin."+Utils.config(key+".broker.type")).newInstance();
			mqs.put(key, mq);
			return mq;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}
	
	
	public static abstract class Listener{
		public abstract <T> void execute(String message);
	}
	public static void on(Class cls,Trigger.Index index,final Notify.Listener listener,Trigger.Type... types){
		List<String> destinationNames = new ArrayList();
		for(Trigger.Type type:types){
			destinationNames.add(cls.getSimpleName()+"."+index.name()+"."+type.name());
			Notify.addSubscribe(cls.getSimpleName()+"."+index.name()+"."+type.name());
		}
		getMQ(DatabaseNames.get(cls, "broker")).addMessageListener(DatabaseNames.get(cls, "broker"),destinationNames.toArray(new String[]{}),listener, false);
	}
	public static void on(Class cls,final Notify.Listener listener,String tag){
		getMQ(DatabaseNames.get(cls, "broker")).addMessageListener(DatabaseNames.get(cls, "broker"),new String[]{cls.getSimpleName()+"."+tag},listener, false);
		Notify.addSubscribe(cls.getSimpleName()+"."+tag);
	}
	public static void on(String flag,final Notify.Listener listener,String tag){
		getMQ(flag).addMessageListener(flag,new String[]{tag},listener, false);
	}
	public static void on(Listener listener, String tag) {
		getMQ(Notify.defaultFlag).addMessageListener(Notify.defaultFlag,new String[]{tag},listener, false);
	}
	public static void sendTextMessage(String flag,String destinationName, final String message,Boolean isP2P){
		getMQ(flag).sendTextMessage(flag, destinationName, message, isP2P);
	}
	public static void sendMessage(String type,Message message){
		getMQ(Notify.defaultFlag).sendTextMessage(Notify.defaultFlag, type, JSON.toJSONString(message), false);
	}
	
}